package com.ucex.common.rocketmq.producer;

import java.util.Properties;

import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.Producer;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.bean.ProducerBean;
import com.aliyun.openservices.shade.com.alibaba.fastjson.JSON;
import com.ucex.common.rocketmq.MQProperties;
import com.ucex.common.rocketmq.domain.MessageEntity;
import com.ucex.common.rocketmq.producer.ali.AbstractMessageProducerProcess;

import lombok.extern.slf4j.Slf4j;

/**
 * 阿里MQ生产者
 *
 */
@Slf4j
public class AliMQProducerProcess extends AbstractMessageProducerProcess {


    private Producer producer;

    private static volatile AliMQProducerProcess INSTANCE;
    private AliMQProducerProcess() {
    }
    public static AliMQProducerProcess getInstance() {
        if (INSTANCE == null) {
            synchronized (AliMQProducerProcess.class) {
                if (INSTANCE == null) {
                    INSTANCE = new AliMQProducerProcess();
                }
            }
        }
        return INSTANCE;
    }

    /**
     * 初始化AliMQ配置信息
     */
    @Override
    public void initProducer(MQProperties mqProperties) {
        Properties properties = new Properties();
        properties.put(PropertyKeyConst.AccessKey, mqProperties.getAccessKey());
        properties.put(PropertyKeyConst.SecretKey, mqProperties.getSecretKey());
        properties.put(PropertyKeyConst.NAMESRV_ADDR, mqProperties.getNameSrvAddr());
        properties.put(PropertyKeyConst.SendMsgTimeoutMillis, 10000);
        producer = ONSFactory.createProducer(properties);
        producer.start();
    }

    @Override
    public void sendMessageFromQueue() {
        Message message = null;
        try {
            message = pool.take();
            producer.send(message);
            log.info("send message to quene topic is: {}, body is: {}",message.getTopic(), new String(message.getBody()));
        } catch (InterruptedException e) {
            log.error("[{}] is stopping, due to following error:", AliMQProducerProcess.class);
            log.error(e.getMessage(), e);
            Thread.interrupted();
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }

    @Override
    public void pushMessageToQueue(MessageEntity msg) {
        try {
            String json = JSON.toJSONString(msg);
            if (null == json || json.trim().isEmpty()) {
                return;
            }
            String key = msg.getKey();
            if (null == key || key.trim().isEmpty()) {
                key = "";
            }
            Message message = new Message(msg.getTopic(), msg.getTag(), key, json.getBytes());
            //设置消息投送时间，最大为7天
            if (msg.getStartDeliverTime() != null && msg.getStartDeliverTime() > 0l) {
                msg.setStartDeliverTime(msg.getStartDeliverTime());
            }
            pool.offer(message);
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }
}
